package com.bw.gmall.realtime.common.base;

import com.bw.gmall.realtime.common.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;

public abstract class BaseApp {


    // 没有具体实现
    public abstract void handle(StreamExecutionEnvironment env,
                                DataStreamSource<String> stream);
    public void start(int port,int parallelism,String ckAndGroupId,String topic) throws Exception {
        //1. 创建流环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);



        System.setProperty("HADOOP_USER_NAME", "hadoop");
        // 2. 设置并行度
        // 在代码中，在算子设置、在配置文件设置 在参数中设置
        env.setParallelism(parallelism);
        // 3.状态后端及检查点相关配置
        // 3.1 设置状态后端
        env.setStateBackend(new HashMapStateBackend());

        // 3.2 开启 checkpoint
        env.enableCheckpointing(5000);
        // 3.3 设置 checkpoint 模式: 精准一次
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 3.4 checkpoint 存储
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall2023/stream/" + ckAndGroupId);
        // 3.5 checkpoint 并发数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 3.6 checkpoint 之间的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        // 3.7 checkpoint  的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        // 3.8 job 取消时 checkpoint 保留策略
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);

        // 4.消费kafka
        KafkaSource<String> kafkaSource = FlinkSourceUtil.getKafkaSource(ckAndGroupId, topic);
        DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        // 5。打印
//        stream.print();
        handle(env,stream);

        // 6、执行
        env.execute();
    }

}
